package net.xuzhen97.distributed.transaction.order.service.executor;

import cn.hutool.core.util.IdUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import net.xuzhen97.distributed.transaction.order.common.exception.CheckException;
import net.xuzhen97.distributed.transaction.order.common.util.JacksonUtil;
import net.xuzhen97.distributed.transaction.order.dao.*;
import net.xuzhen97.distributed.transaction.order.service.dto.OrderPayCompleteConsistencyCmd;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * OrderPayCompleteConsistencyCmd 执行执行器
 * rocketmq 事务消息一致性测试
 * @author xuzhen97
 */
@Service
@Slf4j
public class OrderPayCompleteConsistencyCmdExe implements IRocketMqTransactionExe<OrderPayCompleteConsistencyCmd, String>{

    /**
     * Business 业务标识
     */
    public static final String BUSINESS_ID = "pay-complete";
    /**
     * 业务使用topic
     */
    public static final String ROCKET_MQ_TOPIC = "tx_order";

    private final OrderRepository orderRepository;
    private final RocketmqUndoLogRepository rocketmqUndoLogRepository;
    private final RocketMQTemplate rocketMQTemplate;

    @Autowired
    public OrderPayCompleteConsistencyCmdExe(OrderRepository orderRepository
            , RocketMQTemplate rocketMQTemplate
            , RocketmqUndoLogRepository rocketmqUndoLogRepository){
        this.orderRepository = orderRepository;
        this.rocketMQTemplate = rocketMQTemplate;
        this.rocketmqUndoLogRepository = rocketmqUndoLogRepository;
    }

    /**
     * 执行器逻辑执行
     * @param cmd
     * @return
     */
    public String execute(OrderPayCompleteConsistencyCmd cmd){
        cmd.validate();
        return prepare(cmd);
    }

    /**
     * 预处理阶段，检验数据，给rocketmq发送半消息
     * @param cmd
     * @return
     */
    @Override
    public String prepare(OrderPayCompleteConsistencyCmd cmd){
        cmd.validate();

        Order order = orderRepository.getById(cmd.getId());

        // 不存在的订单抛出异常
        if(order == null){
            throw new CheckException("订单不存在!");
        }

        // 只有存在的订单以及未支付的才会走接下来的逻辑
        if(Objects.equals(OrderState.UNPAY.name(), order.getState())){
            log.info("准备发送事件！！！");
            MessageParams messageParams = new MessageParams();
            messageParams.setCourseId(order.getCourseId());
            messageParams.setUserId(order.getUserId());
            messageParams.setOpUserId(cmd.getOpUserId());

            String trxId = IdUtil.getSnowflakeNextIdStr();

            messageParams.setTrxId(trxId);

            // 发送事务消息，需要生成事务id放到header
            Message<MessageParams> message = MessageBuilder.withPayload(messageParams)
                    .setHeader(RocketMQHeaders.TRANSACTION_ID, trxId)
                    .setHeader("BUSINESS_ID", BUSINESS_ID)
                    .build();

            rocketMQTemplate.sendMessageInTransaction(null
                    ,OrderPayCompleteConsistencyCmdExe.ROCKET_MQ_TOPIC + ":" + OrderPayCompleteConsistencyCmdExe.BUSINESS_ID
                    , message, cmd);

            return order.getId();
        }
        return order.getId();
    }

    /**
     * 本地事务提交阶段，rocketmq 事务监听调用
     *  代码仍然需要检查数据完整性
     * @param cmd
     * @param trxId
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public RocketMQLocalTransactionState commit(OrderPayCompleteConsistencyCmd cmd, String trxId){

        Order order = orderRepository.getById(cmd.getId());

        // 不存在的订单, 返回回滚状态，让rocketmq回滚事务消息
        if(order == null){
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        // 只有存在的订单以及未支付的才会走接下来的逻辑
        if(Objects.equals(OrderState.UNPAY.name(), order.getState())){

            String oldData = JacksonUtil.toJson(order);

            LocalDateTime now = LocalDateTime.now();

            // 修改订单状态未支付状态
            order.setState(OrderState.PAY.name());
            order.setLastModifiedBy(cmd.getOpUserId());
            order.setLastModifiedDate(now);

            String newData = JacksonUtil.toJson(order);

            orderRepository.updateById(order);

            // 增加rocketmq事务消息的回滚日志
            RocketmqUndoLog undoLog = new RocketmqUndoLog();
            undoLog.setTrxId(trxId);
            undoLog.setBusiness(BUSINESS_ID);
            undoLog.setOldData(oldData);
            undoLog.setNewData(newData);
            undoLog.setRollback(false);

            undoLog.setCreatedBy(cmd.getOpUserId());
            undoLog.setCreatedDate(now);
            undoLog.setLastModifiedBy(cmd.getOpUserId());
            undoLog.setLastModifiedDate(now);

            undoLog.setRemarks("支付成功回调事务");

            rocketmqUndoLogRepository.save(undoLog);

            return RocketMQLocalTransactionState.COMMIT;
        }

        // 订单已经支付过了，但是存在半消息，返回中间状态，需要事务消息去走check逻辑
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    /**
     * rocketmq 事务消息回滚逻辑
     * @param trxId
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void rollback(String trxId){
        RocketmqUndoLog undoLog = rocketmqUndoLogRepository.getById(trxId);
        // 回滚日志存在，而且没有回滚过的才可以进行回滚逻辑
        if(undoLog != null && !undoLog.getRollback()){
            //TODO 按照实际业务实现

        }
    }

    @Override
    public String getBusinessId() {
        return BUSINESS_ID;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState check(String trxId) {
        RocketmqUndoLog undoLog = rocketmqUndoLogRepository.getById(trxId);
        if(undoLog == null || undoLog.getRollback()){
            return RocketMQLocalTransactionState.ROLLBACK;
        }else {
            return RocketMQLocalTransactionState.COMMIT;
        }
    }

    @Data
    public static class MessageParams{
        private String trxId;
        private String userId;
        private String courseId;
        private String opUserId;
    }

}
